use sq_global::{API_STOCK_COMP, API_STOCK_DAILY, DATE_FORMAT};
use anyhow::anyhow;
use chrono::Local;
use polars::prelude::{IntoLazy, SortMultipleOptions, col};
use tokio::try_join;
use sq_polars::{get_col_idx, get_str_col};

pub async fn run() -> anyhow::Result<Vec<String>> {
    let curr = Local::now().naive_local();
    let curr = curr.format(DATE_FORMAT);
    let params = vec![("trade_date", curr.to_string())];

    let (comp_df, daily_df) = try_join!(
        sq_tushare::load_data(API_STOCK_COMP, Vec::new()),
        sq_tushare::load_data(API_STOCK_DAILY, params)
    )?;

    if comp_df.is_empty() || daily_df.is_empty() {
        return Err(anyhow!("empty comp_df or daily_df!"));
    }

    let sort_opt = SortMultipleOptions::new().with_order_descending(true);

    let agg_frame = comp_df
        .lazy()
        .inner_join(daily_df.lazy(), "ts_code", "ts_code")
        .filter(col("pct_chg").is_not_null())
        .sort(["pct_chg"], sort_opt)
        .group_by(["exchange"])
        .agg([
            col("ts_code").first().alias("max_ts_code"),
            col("com_name").first().alias("max_com_name"),
            col("pct_chg").first().alias("max_pct_chg"),
            col("ts_code").last().alias("min_ts_code"),
            col("com_name").last().alias("min_com_name"),
            col("pct_chg").last().alias("min_pct_chg"),
        ]);

    let df = agg_frame.collect()?;

    tracing::info!("join frame {:?}", df);

    let map = get_col_idx!(&df);
    let exchange_idx = map["exchange"];
    let max_ts_code_idx = map["max_ts_code"];
    let max_com_name_idx = map["max_com_name"];
    let max_pct_chg_idx = map["max_pct_chg"];

    let min_ts_code_idx = map["min_ts_code"];
    let min_com_name_idx = map["min_com_name"];
    let min_pct_chg_idx = map["min_pct_chg"];

    let mut msg_vec = Vec::new();

    for idx in 0..df.height() {
        let row = df.get_row(idx)?;
        let exchange_val = get_str_col!(row, exchange_idx);

        let exchange = match exchange_val.as_ref() {
            "SSE" => "上交所",
            "SZSE" => "深交所",
            "BSE" => "北交所",
            _ => "未知交易所",
        };

        let max_ts_code_val = get_str_col!(row, max_ts_code_idx);
        let max_com_name_val = get_str_col!(row, max_com_name_idx);
        let max_pct_chg_val = get_str_col!(row, max_pct_chg_idx);
        let min_ts_code_val = get_str_col!(row, min_ts_code_idx);
        let min_com_name_val = get_str_col!(row, min_com_name_idx);
        let min_pct_chg_val = get_str_col!(row, min_pct_chg_idx);
        
        let msg = format!(
            "快讯！：\n
        {exchange} 上市公司{max_com_name_val}（股票代码{max_ts_code_val}）今日上涨最多，达到{max_pct_chg_val}%。
        \n{min_com_name_val}（股票代码{min_ts_code_val}）今日下跌最多，达到{min_pct_chg_val}%。"
        );
        msg_vec.push(msg);
    }

    Ok(msg_vec)
}
